package com.message.source;

import com.message.domain.MessageSourceDto;
import com.message.util.DingdingUtil;
import com.message.util.DruidUtils;
import com.message.util.ReflectUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;


/**
 * 查询最近30分钟数据
 */
@Slf4j
public class MessageSource extends RichSourceFunction<MessageSourceDto> {
    private static final CountDownLatch countDownLatch = new CountDownLatch(1);
    private static final Logger logger = LoggerFactory.getLogger(MessageSource.class);
    private static final long serialVersionUID = 1L;
    private Connection connection;
    private PreparedStatement preparedStatement;
    private Boolean is_run = true;
    private static String sql = "SELECT T\n" +
            "\t.*,\n" +
            "\tP.NAME AS project_name,\n" +
            "\tw.subject AS wbs_subject,\n" +
            "\tfu.lastname AS fromuser,\n" +
            "\ttu.lastname AS touser,\n" +
            "\ttu.LOGIN AS toaccount \n" +
            "FROM\n" +
            "\tPUBLIC.notifications\n" +
            "\tT LEFT JOIN PUBLIC.projects P ON T.project_id = P.\n" +
            "\tID INNER JOIN PUBLIC.work_packages w ON T.resource_type = 'WorkPackage' \n" +
            "\tAND T.resource_id = w.\n" +
            "\tID INNER JOIN PUBLIC.users fu ON T.actor_id = fu.\n" +
            "\tID INNER JOIN PUBLIC.users tu ON T.recipient_id = tu.ID \n" +
            "\tWHERE T.created_at >= (current_timestamp - interval '11 min' )" +
            "ORDER BY\n" +
            "\t updated_at DESC";

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        connection = DruidUtils.getConnection();
        preparedStatement = connection.prepareStatement(sql);
        super.open(parameters);
        log.info("执行MessageSource的open方法");
    }

    @Override
    public synchronized void run(SourceContext<MessageSourceDto> sourceContext) throws Exception {
        MessageSourceDto messageSourceDto = null;
        while (is_run) {
            ResultSet resultSet = preparedStatement.executeQuery();
            String accessToken = "";
            while (resultSet.next()) {
                if("".equals(accessToken)){
                    accessToken=DingdingUtil.getAccessToken();//如果有数据，再获取钉钉凭证
                }
                messageSourceDto = new MessageSourceDto();
                ReflectUtil.datafuzhi(resultSet, messageSourceDto);
                messageSourceDto.setToken(accessToken);
                sourceContext.collect(messageSourceDto);
            }
            countDownLatch.await(10, TimeUnit.MINUTES);
            accessToken = "";
        }
    }

    @Override
    public void cancel() {
        this.is_run = false;
        try {
            super.close();
            if (connection != null) {
                connection.close();
            }
            if (preparedStatement != null) {
                preparedStatement.close();
            }
        } catch (Exception e) {
            logger.error("runException:{}", e);
        }
    }

    private MessageSource() {
    }

    private static class Sigleton {
        private static final MessageSource messageSource = new MessageSource();
    }

    public static MessageSource getInstance() {
        return Sigleton.messageSource;
    }

}
